package kassiopeia.agent.seda.stage;

import kassiopeia.agent.seda.config.State;
import kassiopeia.agent.seda.event.EventHandlerFactory;
import kassiopeia.agent.seda.event.EventQueue;
import kassiopeia.agent.seda.threads.StagePool;
import kassiopeia.plugin.Event;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Configurable;

import java.util.concurrent.TimeUnit;


@Configurable(dependencyCheck = true)
public class ProcessorExecutor implements PluginExecutor {

    private Logger logger = Logger.getLogger(ProcessorExecutor.class);

    private String pluginInstanceId;
    private EventQueue input;
    private Controller controller;
    private Stage stage;
    private EventHandlerFactory eventHandlerFactory;

    public ProcessorExecutor(Stage stage) {
        this.input = stage.getEventQueue();
        this.controller = stage.getController();
        this.stage = stage;
        this.pluginInstanceId = stage.getPlugin().getContext().getPluginInstanceId();
        eventHandlerFactory = new EventHandlerFactory(stage.getPlugin());
    }

    @Override
    public void execute() {
        logger.info("Starting processor plugin: " + pluginInstanceId);

        StagePool stagePool = new StagePool();
        controller.setStagePool(stagePool);

        runController();
        runProcessor(stagePool);

        stopProcessor(stagePool);
        stopController();

        logger.info("Processor plugin: " + pluginInstanceId + " finished");
    }

    private void runController() {
         new Thread(controller).start();
    }

    private void runProcessor(StagePool stagePool) {
        while (stage.getState() == State.RUNNING) {
            try {
                while (!(stagePool.activeCount() < stagePool.size())) {
                    if (stage.getState() != State.RUNNING) {
                        break;
                    }
                    try {
                        TimeUnit.MILLISECONDS.sleep(500);
                    } catch (InterruptedException ex) {
                        logger.error(ex, ex);
                    }
                }
                Event event = input.poll(5L, TimeUnit.SECONDS);
                if (event != null) {
                    stagePool.execute(eventHandlerFactory.getEventHandler(event));
                }
            } catch (InterruptedException ex) {
                logger.error(ex, ex);
            }
        }
    }

    private void stopProcessor(StagePool stagePool) {
        logger.info("Stopping processor " + pluginInstanceId + "...");
        stagePool.shutdown();
        logger.info("Processor " + pluginInstanceId + " stopped");
    }

    private void stopController() {
        logger.info("Stopping controller " + pluginInstanceId + "...");
        controller.stop();
        while (controller.getState() != State.STOPPED) {
            try {
                TimeUnit.SECONDS.sleep(1L);
            } catch (InterruptedException ex) {
                logger.error(ex, ex);
            }
        }
        logger.info("Controller " + pluginInstanceId + " stopped");
    }
}
